package com.haizhi.task;

import com.dinstone.beanstalkc.BeanstalkClient;
import com.haizhi.beanstalk.BeanstalkFactory;
import com.haizhi.config.BeanstalkConf;
import com.haizhi.enums.CrawlStatus;
import com.haizhi.enums.SearchStatus;
import com.haizhi.gsxt.CrawlSchema;
import com.haizhi.gsxt.GenRedisKey;
import com.haizhi.gsxt.ProvinceTable;
import com.haizhi.model.CrawlRequest;
import com.haizhi.mongo.ScanIter;
import com.haizhi.redis.RedisHandler;
import com.haizhi.util.JsonUtil;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoDatabase;
import com.mongodb.client.model.IndexOptions;
import org.apache.commons.lang3.StringUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.bson.Document;
import redis.clients.jedis.Jedis;

import java.util.ArrayList;
import java.util.List;

public class SchedulerWorker {

    private static final Logger logger = LogManager.getLogger(SchedulerWorker.class);

    //最大调度时间 秒
    private static final int MAX_SCHEDULER_SECONDS = 10 * 60;

    //当前省份
    private String province;

    //消息队列阻塞阈值
    private int beanstalkThresholdNum;

    //消息队列生产者
    private BeanstalkClient producer;
    private String beanstalkTube;

    //MongoDB句柄
    private MongoCollection<Document> collection;

    //redis 句柄
    private Jedis jedis;

    /**
     * redis 句柄
     */
    private RedisHandler redisHandler;

    /**
     * mongodb 游标
     */
    private ScanIter firstIter;
    private ScanIter secondIter;

    public SchedulerWorker(String province,
                           int beanstalkThresholdNum,
                           BeanstalkConf beanstalkConf,
                           MongoDatabase mongoDatabase, RedisHandler redisHandler) {

        //省份 拼音的
        this.province = province;

        //消息队列阻塞阈值
        this.beanstalkThresholdNum = beanstalkThresholdNum;


        //初始化redis
        this.redisHandler = redisHandler;
        this.jedis = redisHandler.getJedisObject();

        //初始化MongoDB
        this.collection = mongoDatabase.getCollection(ProvinceTable.getProvinceTable(province));

        //初始化消息队列
        this.producer = BeanstalkFactory.getProducer(beanstalkConf.getHost(),
                beanstalkConf.getPort(),
                beanstalkConf.getTube());
        beanstalkTube = beanstalkConf.getTube();

        //创建索引
        createIndex();

        //初始化两个游标
        initScanIter();
    }

    /**
     * 创建索引
     */
    private void createIndex() {
        collection.createIndex(new Document(CrawlSchema.CRAWL_STATUS, 1), new IndexOptions().background(true));
        collection.createIndex(new Document(CrawlSchema.SEARCH_STATUS, 1), new IndexOptions().background(true));
    }

    private void initScanIter() {
        Document filter = new Document();

        Document crawlStatusNone = new Document(CrawlSchema.CRAWL_STATUS, null);
        Document crawlStatusFail = new Document(CrawlSchema.CRAWL_STATUS, CrawlStatus.CRAWL_DETAIL_FAIL);

        List<Document> crawlList = new ArrayList<>();
        crawlList.add(crawlStatusFail);
        crawlList.add(crawlStatusNone);
        Document or = new Document("$or", crawlList);

        Document searchStatus = new Document(CrawlSchema.SEARCH_STATUS, new Document("$ne", SearchStatus.SEARCH_LIST_NOT_FOUND));

        List<Document> allList = new ArrayList<>();
        allList.add(or);
        allList.add(searchStatus);

        filter.append("$and", allList);

        firstIter = new ScanIter(collection, filter);


        //初始化第二个游标
        allList = new ArrayList<>();
        allList.add(new Document(CrawlSchema.CRAWL_STATUS, CrawlStatus.CRAWL_DETAIL_PART_SUCCESS));
        allList.add(new Document(CrawlSchema.SEARCH_STATUS, new Document("$ne", SearchStatus.SEARCH_LIST_NOT_FOUND)));
        filter = new Document("$and", allList);

        secondIter = new ScanIter(collection, filter);
    }

    //获得调度种子
    private Document getScheduleSeed() {
        Document seed = firstIter.getNextDocument();
        if (seed != null) {
            return seed;
        }
        return secondIter.getNextDocument();
    }

    //获得调度请求结构体
    private CrawlRequest getScheduleRequest(Document seed) {
        if (null == seed) {
            return null;
        }

        CrawlRequest request = new CrawlRequest();
        String company = seed.getString(CrawlSchema.ID);
        if (StringUtils.isEmpty(company)) {
            return null;
        }

        request.setCompany(company);
        request.setProvince(province);
        request.setScheduler(true);

        return request;

    }

    // 调度
    public boolean schedule() {
        try {
            //先判断下游消息队列是否支持发送
            long readyJobsNum = BeanstalkFactory.getReadyJobsNum(producer, beanstalkTube);
            if (readyJobsNum > beanstalkThresholdNum) {
                logger.warn("消息队列阻塞: province = {} num = {}", province, readyJobsNum);
                return false;
            }

            //获得当前待发送种子
            Document seed = getScheduleSeed();
            if (null == seed) {
                logger.warn("当前没有适合调度的种子: province = {}", province);
                return false;
            }
            //转换成request
            CrawlRequest request = getScheduleRequest(seed);
            if (null == request) {
                logger.error("种子转换失败: province = {} seed = {}", province, seed.toJson());
                return false;
            }

            //判断种子是否已经在调度缓存
            String scheduleKey = GenRedisKey.genScheduleKey(request.getProvince(), request.getCompany());
            if (redisHandler.isCacheExist(scheduleKey, jedis)) {
                logger.warn("当前种子已经正在调度: {} : {} : {}", scheduleKey, province, request.getCompany());
                return false;
            }

            String jsonStr = JsonUtil.objectToJson(request);
            if (null == jsonStr) {
                logger.error("请求转换json失败: province = {} request = {}", province, request);
                return false;
            }

            //发送种子
            BeanstalkFactory.put(producer, jsonStr);

            //把种子信息加入调度缓存
            redisHandler.saveCacheEx(GenRedisKey.genScheduleKey(
                    request.getProvince(), request.getCompany()),
                    GenRedisKey.getRedisValue(request.getProvince(), request.getCompany()),
                    MAX_SCHEDULER_SECONDS, jedis);

            logger.info("调度成功: {} : {} : {}", scheduleKey, province, request.getCompany());
        } catch (Exception e) {
            logger.error("调度异常: province = {}", province, e);
            return false;
        }
        return true;
    }

    public void close() {
        if (null != firstIter) {
            firstIter.close();
        }
        if (null != secondIter) {
            secondIter.close();
        }
        redisHandler.returnJedisOjbect(jedis);
        producer.close();
        logger.info("资源释放完成: {}", province);
    }
}
